import numpy as np
import pandas as pd
from mysql_client import MySQLSSHClient
from custom_logger import PrefixDateLogger

logger = PrefixDateLogger("monitor_dau_dnu")

class DauDnuMonitor:
    def __init__(self):
        self.client = MySQLSSHClient()
        self.client.connect()

    def get_dau_dnu_data(self):
        """查询最近10天的DAU/DNU数据"""
        sql = """
        SELECT DATE, `user_id`, 
               SUM(dau) AS dau_new, 
               SUM(wau) AS wau_new, 
               SUM(mau) AS mau_new, 
               SUM(dnu) AS dnu_new, 
               SUM(uv) AS uv_new 
        FROM cfx_t.data_cpa_user_dau_2 
        WHERE DATE >= UNIX_TIMESTAMP(DATE_SUB(NOW(), INTERVAL 10 DAY)) * 1000 
        GROUP BY DATE , user_id
        ORDER BY DATE, user_id
        """
        try:
            # 使用query方法而不是read_sql
            data, columns = self.client.query(sql)
            # 将查询结果转换为DataFrame
            df = pd.DataFrame(data, columns=columns)
            
            # 将所有数值列转换为float类型（解决Decimal类型问题）
            numeric_columns = ['dau_new', 'wau_new', 'mau_new', 'dnu_new', 'uv_new']
            for col in numeric_columns:
                df[col] = pd.to_numeric(df[col], errors='coerce')
            
            logger.info(f"查询到 {len(df)} 条数据")
            logger.info(f"列名: {columns}")
            logger.info(f"前几行数据: {df.head()}")
            return df
        except Exception as e:
            logger.error(f"查询数据失败: {str(e)}")
            raise

    def analyze_data(self, df):
        """
        分析数据，按user_id比较昨天数据与前三天平均值
        返回：需要告警的数据列表
        """
        if df.empty:
            logger.warning("没有查询到数据")
            return []
        
        # 排除user_id为-1的数据
        df = df[df['user_id'] != -1].copy()
        logger.info(f"排除user_id=-1后，剩余 {len(df)} 条数据")

        # 转换DATE为日期格式（从毫秒时间戳）
        # 确保DATE列是数值类型
        df['DATE'] = pd.to_numeric(df['DATE'], errors='coerce')
        # 转换为UTC时间，然后转为东八区（Asia/Shanghai）
        df['date_formatted'] = pd.to_datetime(df['DATE'], unit='ms', utc=True).dt.tz_convert('Asia/Shanghai').dt.normalize()
        logger.info(f"前几行数据: {df.head()}")
        
        logger.info(f"查询到的日期范围: {df['date_formatted'].min()} 到 {df['date_formatted'].max()}")
        logger.info(f"共 {len(df)} 条数据")
        
        # 获取昨天的日期（排除今天）
        # 使用Asia/Shanghai时区，与date_formatted保持一致
        today = pd.Timestamp.now(tz='Asia/Shanghai').normalize()
        yesterday = today - pd.Timedelta(days=1)
        
        logger.info(f"今天日期: {today.strftime('%Y-%m-%d')}")
        logger.info(f"昨天日期: {yesterday.strftime('%Y-%m-%d')}")
        
        # 过滤出昨天及之前的数据（排除今天的不完整数据）
        historical_data = df[df['date_formatted'] < today].copy()
        
        if historical_data.empty:
            logger.warning("没有历史数据")
            return []
        
        # 获取所有日期列表
        unique_dates = sorted(historical_data['date_formatted'].unique())
        
        if len(unique_dates) < 4:
            logger.warning(f"历史数据不足4天，无法进行对比分析。当前历史数据天数: {len(unique_dates)}")
            return []
        
        logger.info(f"历史数据日期: {[d.strftime('%Y-%m-%d') for d in unique_dates]}")
        
        # 获取昨天的日期和前三天的日期
        yesterday_date = unique_dates[-1]
        previous_3_dates = unique_dates[-4:-1]
        
        logger.info(f"昨天数据日期: {yesterday_date.strftime('%Y-%m-%d')}")
        logger.info(f"前3天数据日期: {[d.strftime('%Y-%m-%d') for d in previous_3_dates]}")
        
        # 获取昨天的数据
        yesterday_data = historical_data[historical_data['date_formatted'] == yesterday_date].copy()
        logger.info(f"前几行数据: {yesterday_data}")
        
        # 获取前三天的数据
        previous_3_days_data = historical_data[historical_data['date_formatted'].isin(previous_3_dates)].copy()
        
        # 按user_id计算前三天的平均值（只计算dau和dnu）
        avg_3_days = previous_3_days_data.groupby('user_id').agg({
            'dau_new': 'mean',
            'dnu_new': 'mean'
        }).reset_index()
        
        # 将昨天数据和前三天平均值合并
        comparison_data = yesterday_data.merge(
            avg_3_days, 
            on='user_id', 
            how='inner',
            suffixes=('_yesterday', '_avg3days')
        )
        
        logger.info(f"需要对比的user_id数量: {len(comparison_data)}")
        
        # 检查告警条件
        alerts = []
        metrics = ['dau_new', 'dnu_new']  # 只监控DAU和DNU
        metric_names = {
            'dau_new': 'DAU',
            'dnu_new': 'DNU'
        }
        
        for _, row in comparison_data.iterrows():
            user_id = row['user_id']
            
            for metric in metrics:
                yesterday_value = row[f'{metric}_yesterday']
                avg_value = row[f'{metric}_avg3days']
                
                if avg_value == 0:
                    continue
                
                # 计算变化比例（正值表示下降，负值表示上升）
                change_percent = (avg_value - yesterday_value) / avg_value * 100
                
                # 如果变化超过20%（上升或下降），添加到告警列表
                if abs(change_percent) > 20:
                    change_type = "下降" if change_percent > 0 else "上升"
                    alerts.append({
                        'user_id': user_id,
                        'metric': metric_names[metric],
                        'yesterday_value': yesterday_value,
                        'avg_value': avg_value,
                        'change_percent': abs(change_percent),
                        'change_type': change_type,
                        'yesterday_date': yesterday_date.strftime('%Y-%m-%d')
                    })
                    logger.info(f"告警: user_id={user_id}, {metric_names[metric]}: 昨天={yesterday_value}, 前3天平均={avg_value:.2f}, {change_type}比例={abs(change_percent):.2f}%")
        
        return alerts, comparison_data, yesterday_date

    def generate_report_table(self, comparison_data):
        """生成数据对比表格HTML"""
        # 只显示前10个user_id的对比数据
        display_data = comparison_data.head(10).copy()
        
        # 选择要显示的列并重命名
        if not display_data.empty:
            display_data = display_data[['user_id', 'dau_new_yesterday', 'dau_new_avg3days', 
                                        'dnu_new_yesterday', 'dnu_new_avg3days']].copy()
            # 重命名列为中文
            display_data.columns = ['User ID', '昨天活跃', '前三天平均活跃', '昨天新增', '前三天平均新增']
            # 将数值列转换为整数
            for col in ['昨天活跃', '前三天平均活跃', '昨天新增', '前三天平均新增']:
                display_data[col] = display_data[col].astype(int)
        
        return display_data.to_html(index=False)

    def generate_alert_content(self, alerts):
        """生成告警邮件内容"""
        if not alerts:
            return None
        
        alert_list_html = "<ul>"
        for alert in alerts:
            change_type = alert['change_type']
            color = "red" if change_type == "下降" else "green"
            alert_list_html += f"""
            <li>
                <strong>User ID: {alert['user_id']}</strong> - <strong>{alert['metric']}</strong>: 
                昨天 {alert['yesterday_value']:.0f}, 
                前3天平均 {alert['avg_value']:.2f}, 
                <span style="color:{color};">{change_type} {alert['change_percent']:.2f}%</span>
            </li>
            """
        alert_list_html += "</ul>"
        
        return alert_list_html

    def close(self):
        """关闭数据库连接"""
        if self.client:
            self.client.close()


# 使用示例
if __name__ == "__main__":
    monitor = DauDnuMonitor()
    try:
        df = monitor.get_dau_dnu_data()
        result = monitor.analyze_data(df)
        
        if result:
            alerts, comparison_data, yesterday_date = result
            
            if alerts:
                logger.info(f"发现 {len(alerts)} 个告警指标")
                alert_content = monitor.generate_alert_content(alerts)
                print("告警内容：")
                print(alert_content)
                
                table_html = monitor.generate_report_table(comparison_data)
                print("\n数据表格：")
                print(table_html)
            else:
                logger.info("所有指标正常，无需告警")
    finally:
        monitor.close()
